package com.jiujichina.star.producer;

import kafka.producer.KeyedMessage;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Properties;
import java.util.Random;
import java.util.UUID;

/**
 * 生产者
 * 使用@EnableScheduling注解开启定时任务
 */
@Component
@EnableScheduling
public class KafkaProducerDemo implements CommandLineRunner {

    Logger logger = LoggerFactory.getLogger(KafkaProducerDemo.class);

    @Autowired
    private KafkaProducer producer;

    private final  static String TOPIC = "test";


  /*  @Scheduled(cron = "0/10 * * * * ?")
    public void send(){
        String topic = "test";
        String uuid = UUID.randomUUID().toString();
        //消息的Key
        String messageKey = uuid;
        String messageValue = "Message_" + uuid + new Random(10).nextInt();
        long startTime = System.currentTimeMillis();
        ProducerRecord<Integer, String> producerRecord = new ProducerRecord(topic, messageKey, messageValue);
        try {
            producer.send(producerRecord).get();
            logger.info("发送消息{}",messageValue);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }*/

    public void sendMsg()throws InterruptedException{
        String uuid = UUID.randomUUID().toString();
        //消息的Key
        ProducerRecord<Integer, String> message1 = new ProducerRecord(TOPIC, "31", "test 31");
        producer.send(message1);
        Thread.sleep(5000);
        ProducerRecord<Integer, String> message2 = new ProducerRecord(TOPIC, "31", "test 31");
        producer.send(message2);
        Thread.sleep(5000);
        ProducerRecord<Integer, String> message3 = new ProducerRecord(TOPIC, "32", "test 32");
        producer.send(message3);
    }



    @Override
    public void run(String... strings) throws Exception {
        new Thread(() -> {
            try {
                sendMsg();
            }catch (Exception e){e.printStackTrace();}
        }).start();
    }


}
